package org.kf

import java.util.Properties

import org.FlinkStreamApp
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.kafka.clients.consumer.ConsumerConfig

/**
 * description ：flink 读取 kafka　数据
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/25 18:14
 * modified By ：
 * version:    : 1.0
 */
object KafkaRead extends FlinkStreamApp {
  override def doSomeThing(environment: StreamExecutionEnvironment) = {
    // 创建 kafka 连接配置文件
    val properties = new Properties()
    properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "my-group-01")
    properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop101:9092")
    properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")
    properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")

    // 读取 kafka 数据
    val source = environment.addSource(new FlinkKafkaConsumer[String]("t01", new SimpleStringSchema(), properties))

    // 打印数据
    source.print()
  }
}
